package com.iwhalecloud.uncc.mq.consumer;

import java.util.Properties;

import com.iwhalecloud.uncc.common.SysConstant;
import org.apache.commons.lang.exception.ExceptionUtils;

import com.ctg.mq.api.CTGMQFactory;
import com.ctg.mq.api.IMQPushConsumer;
import com.ctg.mq.api.PropertyKeyConst;
import com.ctg.mq.api.enums.MQMessageModel;
import com.ctg.mq.api.listener.ConsumerQueueListener;
import com.ctg.mq.api.listener.ConsumerTopicListener;
import com.ctg.mq.api.listener.Listener;
import com.iwhalecloud.uncc.model.MqClusterConfig;
import com.iwhalecloud.uncc.model.MqConsumerConfig;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;

@Slf4j
public class CtgMqPushConsumer implements MqConsumer {

	private MqClusterConfig clusterConfig;
	private MqConsumerConfig consumerConfig; // 配置
	private IMQPushConsumer consumer;
	private Listener messageListener;

	public CtgMqPushConsumer(MqClusterConfig mqClusterConfig, MqConsumerConfig consumerConfig,
			Listener messageListener) {
		this.clusterConfig = mqClusterConfig;
		this.consumerConfig = consumerConfig;
		this.messageListener = messageListener;
	}

	@Override
	public void init() {
		log.info("init consumer start, consumerName: " + consumerConfig.getConsumerName());
		try {
			if (consumer == null) {
				Properties p = new Properties();
				p.setProperty(PropertyKeyConst.NamesrvAddr, clusterConfig.getNameserv());
				p.setProperty(PropertyKeyConst.NamesrvAuthID, clusterConfig.getUsername());
				p.setProperty(PropertyKeyConst.NamesrvAuthPwd, clusterConfig.getPwd());
				p.setProperty(PropertyKeyConst.ClusterName, clusterConfig.getClusterName());
				p.setProperty(PropertyKeyConst.TenantID, clusterConfig.getTenantId());
				p.setProperty(PropertyKeyConst.ConsumeThreadMax, String.valueOf(consumerConfig.getConsumeThreadMax()));
				p.setProperty(PropertyKeyConst.ConsumeThreadMin, String.valueOf(consumerConfig.getConsumeThreadMin()));
				p.setProperty(PropertyKeyConst.ClusterConsume,
						SysConstant.MQ_MESSAGE_MODEL_BROADCASTING.equalsIgnoreCase(consumerConfig.getMessageModel()) ?
								MessageModel.BROADCASTING.getModeCN() : MQMessageModel.CLUSTERING.getModeCN());
				p.setProperty(PropertyKeyConst.ConsumeFromWhere, "CONSUME_FROM_LAST_OFFSET");
				p.setProperty(PropertyKeyConst.ConsumerGroupName, consumerConfig.getConsumerName());
				p.setProperty(PropertyKeyConst.ConsumeMessageBatchMaxSize, String.valueOf(consumerConfig.getConsumeBatchSize()));

				consumer = CTGMQFactory.createPushConsumer(p);
			}
			log.info("init consumer success, consumerName: " + consumerConfig.getConsumerName());
		} catch (Exception e) {
			log.info("init consumer exception , consumerName: " + consumerConfig.getConsumerName() + ",Exception:"
					+ e.getMessage());
		}
	}

	@Override
	public void listen() throws Exception {
		try {
			if (consumer == null) {
				init();
			}
			if (messageListener instanceof ConsumerQueueListener) {
				ConsumerQueueListener queueListener = (ConsumerQueueListener) messageListener;
				consumer.listenQueue(consumerConfig.getTopic(), consumerConfig.getTag(), queueListener);
			} else if (messageListener instanceof ConsumerTopicListener) {
				ConsumerTopicListener topicListener = (ConsumerTopicListener) messageListener;
				consumer.listenTopic(consumerConfig.getTopic(), consumerConfig.getTag(), topicListener);
			}
			log.info("listen consumer success, consumerName: " + consumerConfig.getConsumerName() + ",topic:"
					+ consumerConfig.getTopic());
		} catch (Exception e) {
			log.info("listen consumer exception, consumerName: " + consumerConfig.getConsumerName() + ",topic:"
					+ consumerConfig.getTopic() + "," + ExceptionUtils.getStackTrace(e));
		}

	}

	@Override
	public void shutdown() throws Exception {
		if (consumer != null) {
			try {
				consumer.close();
			} catch (Exception e) {
				log.error(ExceptionUtils.getStackTrace(e));
			}
		}
	}
}
